//! 执行算子实现
//!
//! 实现各种查询执行算子，包括表扫描、投影、过滤、连接等

use super::*;
use std::time::Instant;
use std::collections::HashMap;

/// 表扫描算子
/// 从存储引擎中扫描表数据，使用真正的页面式存储
#[derive(Debug)]
pub struct TableScanOperator {
    /// 表名
    table_name: String,
    /// 输出模式
    schema: ExecutionSchema,
    /// 投影列索引（None表示全部列）
    projection: Option<Vec<usize>>,
    /// 当前扫描的行ID
    current_row_id: Option<crate::storage::tuple::RowId>,
    /// 表的行迭代器
    row_iterator: Option<TableRowIterator>,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
    /// 测试数据（用于基准测试）
    test_data: Option<Vec<ExecutionTuple>>,
    /// 存储引擎引用（用于直接访问表数据）
    storage_engine: Option<Arc<parking_lot::RwLock<crate::storage::StorageEngine>>>,
    /// 缓存的表数据
    cached_data: Option<Vec<Vec<crate::sql::Value>>>,
    /// 当前数据索引
    current_index: usize,
}

/// 表行迭代器
/// 用于遍历表中的所有行
#[derive(Debug)]
pub struct TableRowIterator {
    /// 表引用
    table: Arc<crate::storage::Table>,
    /// 当前页面ID
    current_page_id: u64,
    /// 当前槽位ID
    current_slot_id: u16,
    /// 最大事务ID（用于MVCC可见性检查）
    max_transaction_id: u32,
}

impl TableRowIterator {
    /// 创建新的表行迭代器
    pub fn new(
        table: Arc<crate::storage::Table>,
        max_transaction_id: u32,
    ) -> Self {
        Self {
            table,
            current_page_id: 0,
            current_slot_id: 0,
            max_transaction_id,
        }
    }
}

impl Iterator for TableRowIterator {
    type Item = Result<(crate::storage::tuple::RowId, crate::storage::tuple::Tuple)>;

    fn next(&mut self) -> Option<Self::Item> {
        // 简化实现：返回None表示迭代结束
        // 在真实实现中，这里会遍历表的所有页面和槽位
        None
    }
}

/// 投影算子
/// 选择指定的列
pub struct ProjectionOperator {
    /// 输入算子
    input: Box<dyn Operator>,
    /// 投影列索引
    projection_indices: Vec<usize>,
    /// 输出模式
    schema: ExecutionSchema,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
}

/// 过滤算子
/// 根据谓词过滤元组
pub struct FilterOperator {
    /// 输入算子
    input: Box<dyn Operator>,
    /// 过滤谓词
    predicate: FilterPredicate,
    /// 输出模式（与输入相同）
    schema: ExecutionSchema,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
}

/// 哈希连接算子
/// 实现内连接、左连接等
pub struct HashJoinOperator {
    /// 左输入算子（构建端）
    left_input: Box<dyn Operator>,
    /// 右输入算子（探测端）
    right_input: Box<dyn Operator>,
    /// 连接条件
    join_condition: JoinCondition,
    /// 连接类型
    join_type: JoinType,
    /// 哈希表
    hash_table: Option<HashMap<Vec<u8>, Vec<ExecutionTuple>>>,
    /// 输出模式
    schema: ExecutionSchema,
    /// 当前匹配的元组
    current_matches: Vec<ExecutionTuple>,
    /// 当前匹配索引
    current_match_index: usize,
    /// 当前右侧元组
    current_right_tuple: Option<ExecutionTuple>,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
}

/// 过滤谓词
#[derive(Debug, Clone)]
pub enum FilterPredicate {
    /// 比较谓词
    Comparison {
        left_column: usize,
        operator: ComparisonOperator,
        right_value: Value,
    },
    /// 列比较谓词
    ColumnComparison {
        left_column: usize,
        operator: ComparisonOperator,
        right_column: usize,
    },
    /// AND谓词
    And(Box<FilterPredicate>, Box<FilterPredicate>),
    /// OR谓词
    Or(Box<FilterPredicate>, Box<FilterPredicate>),
    /// NOT谓词
    Not(Box<FilterPredicate>),
}

/// 比较操作符
#[derive(Debug, Clone, PartialEq)]
pub enum ComparisonOperator {
    Equal,
    NotEqual,
    LessThan,
    LessThanOrEqual,
    GreaterThan,
    GreaterThanOrEqual,
    Like,
    In,
}

/// 连接条件
#[derive(Debug, Clone)]
pub struct JoinCondition {
    /// 左侧列索引
    pub left_column: usize,
    /// 右侧列索引
    pub right_column: usize,
    /// 比较操作符
    pub operator: ComparisonOperator,
}

/// 连接类型
#[derive(Debug, Clone, PartialEq)]
pub enum JoinType {
    Inner,
    LeftOuter,
    RightOuter,
    FullOuter,
}

impl TableScanOperator {
    /// 创建新的表扫描算子
    pub fn new(
        table_name: String,
        projection: Option<Vec<usize>>,
        context: &ExecutionContext,
    ) -> Result<Self> {
        // 从存储引擎获取表模式
        let table_schema = context.get_table_schema(&table_name)?;

        // 创建执行模式
        let schema = if let Some(ref proj) = projection {
            // 投影指定列
            let columns = proj.iter()
                .map(|&idx| {
                    if idx >= table_schema.columns.len() {
                        return Err(Error::query_execution(format!("列索引 {} 超出范围", idx)));
                    }
                    let col = &table_schema.columns[idx];
                    Ok(ColumnInfo {
                        name: col.name.clone(),
                        data_type: col.data_type.clone(),
                        nullable: col.nullable,
                        table_name: Some(table_name.clone()),
                    })
                })
                .collect::<Result<Vec<_>>>()?;
            ExecutionSchema { columns }
        } else {
            // 全部列
            let columns = table_schema.columns.iter()
                .map(|col| ColumnInfo {
                    name: col.name.clone(),
                    data_type: col.data_type.clone(),
                    nullable: col.nullable,
                    table_name: Some(table_name.clone()),
                })
                .collect();
            ExecutionSchema { columns }
        };

        Ok(Self {
            table_name,
            schema,
            projection,
            current_row_id: None,
            row_iterator: None,
            statistics: OperatorStatistics::default(),
            is_open: false,
            test_data: None,
            storage_engine: None, // 将在open方法中设置
            cached_data: None,
            current_index: 0,
        })
    }

    /// 设置测试数据（用于基准测试）
    pub fn set_test_data(&mut self, data: Vec<ExecutionTuple>) {
        self.test_data = Some(data);
    }

    /// 设置存储引擎引用
    pub fn set_storage_engine(&mut self, storage_engine: Arc<parking_lot::RwLock<crate::storage::StorageEngine>>) {
        self.storage_engine = Some(storage_engine);
    }
}

impl Operator for TableScanOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        let start_time = Instant::now();

        // 如果有存储引擎引用，从中加载数据
        if let Some(ref storage_engine) = self.storage_engine {
            let storage = storage_engine.read();
            if let Some(table) = storage.get_table(&self.table_name) {
                // 使用表的scan_rows方法获取所有数据
                let rows = table.scan_rows(u32::MAX)?; // 使用最大事务ID看到所有数据
                self.cached_data = Some(rows);
                tracing::debug!("从存储引擎加载了 {} 行数据", self.cached_data.as_ref().unwrap().len());
            } else {
                return Err(Error::query_execution(format!("表 {} 不存在", self.table_name)));
            }
        }

        self.current_row_id = None;
        self.current_index = 0;
        self.is_open = true;
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

        tracing::debug!("表扫描算子已打开: {}", self.table_name);
        Ok(())
    }
    
    fn next(&mut self, _context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("算子未打开".to_string()));
        }

        let start_time = Instant::now();

        // 从缓存的数据中获取下一行
        if let Some(ref cached_data) = self.cached_data {
            if self.current_index < cached_data.len() {
                let values = cached_data[self.current_index].clone();
                self.current_index += 1;

                // 应用投影
                let projected_values = if let Some(ref projection) = self.projection {
                    projection.iter()
                        .map(|&idx| {
                            if idx < values.len() {
                                Ok(values[idx].clone())
                            } else {
                                Err(Error::query_execution(format!("投影索引 {} 超出范围", idx)))
                            }
                        })
                        .collect::<Result<Vec<_>>>()?
                } else {
                    values
                };

                // 创建执行元组
                let execution_tuple = ExecutionTuple {
                    values: projected_values,
                    row_id: None, // 简化实现，不提供行ID
                    version_info: None, // 简化实现，不提供版本信息
                };

                // 更新统计信息
                self.statistics.tuples_processed += 1;
                self.statistics.tuples_output += 1;
                self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

                return Ok(Some(execution_tuple));
            }
        }

        // 没有更多数据
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
        Ok(None)
    }
    
    fn close(&mut self, _context: &mut ExecutionContext) -> Result<()> {
        self.is_open = false;
        tracing::debug!("表扫描算子已关闭: {}", self.table_name);
        Ok(())
    }
    
    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }
    
    fn statistics(&self) -> OperatorStatistics {
        self.statistics.clone()
    }
    
    fn reset(&mut self) -> Result<()> {
        self.current_row_id = None;
        self.statistics = OperatorStatistics::default();
        self.is_open = false;
        Ok(())
    }
}

// TableScanOperator的辅助方法已移除，因为现在直接使用缓存数据

impl ProjectionOperator {
    /// 创建新的投影算子
    pub fn new(
        input: Box<dyn Operator>,
        projection_indices: Vec<usize>,
    ) -> Result<Self> {
        let input_schema = input.schema();
        let schema = input_schema.project(&projection_indices)?;
        
        Ok(Self {
            input,
            projection_indices,
            schema,
            statistics: OperatorStatistics::default(),
            is_open: false,
        })
    }
}

impl Operator for ProjectionOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }
        
        self.input.open(context)?;
        self.is_open = true;
        
        tracing::debug!("投影算子已打开");
        Ok(())
    }
    
    fn next(&mut self, context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("算子未打开".to_string()));
        }
        
        let start_time = Instant::now();
        
        if let Some(input_tuple) = self.input.next(context)? {
            // 投影指定列
            let mut projected_values = Vec::new();
            for &index in &self.projection_indices {
                projected_values.push(input_tuple.get_value(index)?.clone());
            }
            
            let projected_tuple = ExecutionTuple {
                values: projected_values,
                row_id: input_tuple.row_id,
                version_info: input_tuple.version_info,
            };
            
            self.statistics.tuples_processed += 1;
            self.statistics.tuples_output += 1;
            self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
            
            Ok(Some(projected_tuple))
        } else {
            self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
            Ok(None)
        }
    }
    
    fn close(&mut self, context: &mut ExecutionContext) -> Result<()> {
        self.input.close(context)?;
        self.is_open = false;
        tracing::debug!("投影算子已关闭");
        Ok(())
    }
    
    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }
    
    fn statistics(&self) -> OperatorStatistics {
        let mut stats = self.statistics.clone();
        let input_stats = self.input.statistics();
        stats.disk_io_count += input_stats.disk_io_count;
        stats.memory_usage_bytes += input_stats.memory_usage_bytes;
        stats
    }
    
    fn reset(&mut self) -> Result<()> {
        self.input.reset()?;
        self.statistics = OperatorStatistics::default();
        self.is_open = false;
        Ok(())
    }
}

impl FilterOperator {
    /// 创建新的过滤算子
    pub fn new(input: Box<dyn Operator>, predicate: FilterPredicate) -> Self {
        let schema = input.schema().clone();

        Self {
            input,
            predicate,
            schema,
            statistics: OperatorStatistics::default(),
            is_open: false,
        }
    }
}

impl Operator for FilterOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        self.input.open(context)?;
        self.is_open = true;

        tracing::debug!("过滤算子已打开");
        Ok(())
    }

    fn next(&mut self, context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("算子未打开".to_string()));
        }

        let start_time = Instant::now();

        loop {
            if let Some(input_tuple) = self.input.next(context)? {
                self.statistics.tuples_processed += 1;

                // 评估谓词
                if self.evaluate_predicate(&input_tuple, &self.predicate)? {
                    self.statistics.tuples_output += 1;
                    self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
                    return Ok(Some(input_tuple));
                }
                // 如果谓词不满足，继续下一个元组
            } else {
                self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
                return Ok(None);
            }
        }
    }

    fn close(&mut self, context: &mut ExecutionContext) -> Result<()> {
        self.input.close(context)?;
        self.is_open = false;
        tracing::debug!("过滤算子已关闭");
        Ok(())
    }

    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }

    fn statistics(&self) -> OperatorStatistics {
        let mut stats = self.statistics.clone();
        let input_stats = self.input.statistics();
        stats.disk_io_count += input_stats.disk_io_count;
        stats.memory_usage_bytes += input_stats.memory_usage_bytes;
        stats
    }

    fn reset(&mut self) -> Result<()> {
        self.input.reset()?;
        self.statistics = OperatorStatistics::default();
        self.is_open = false;
        Ok(())
    }
}

impl FilterOperator {
    /// 评估过滤谓词
    fn evaluate_predicate(&self, tuple: &ExecutionTuple, predicate: &FilterPredicate) -> Result<bool> {
        match predicate {
            FilterPredicate::Comparison { left_column, operator, right_value } => {
                let left_value = tuple.get_value(*left_column)?;
                self.compare_values(left_value, operator, right_value)
            }
            FilterPredicate::ColumnComparison { left_column, operator, right_column } => {
                let left_value = tuple.get_value(*left_column)?;
                let right_value = tuple.get_value(*right_column)?;
                self.compare_values(left_value, operator, right_value)
            }
            FilterPredicate::And(left, right) => {
                Ok(self.evaluate_predicate(tuple, left)? && self.evaluate_predicate(tuple, right)?)
            }
            FilterPredicate::Or(left, right) => {
                Ok(self.evaluate_predicate(tuple, left)? || self.evaluate_predicate(tuple, right)?)
            }
            FilterPredicate::Not(inner) => {
                Ok(!self.evaluate_predicate(tuple, inner)?)
            }
        }
    }

    /// 比较两个值
    fn compare_values(&self, left: &Value, operator: &ComparisonOperator, right: &Value) -> Result<bool> {
        use std::cmp::Ordering;

        // 处理NULL值
        match (left, right) {
            (Value::Null, Value::Null) => {
                return Ok(matches!(operator, ComparisonOperator::Equal));
            }
            (Value::Null, _) | (_, Value::Null) => {
                return Ok(false); // NULL与任何值比较都是false
            }
            _ => {}
        }

        // 比较非NULL值
        let comparison = match (left, right) {
            (Value::Integer(a), Value::Integer(b)) => a.cmp(b),
            (Value::Real(a), Value::Real(b)) => a.partial_cmp(b).unwrap_or(Ordering::Equal),
            (Value::Text(a), Value::Text(b)) => a.cmp(b),
            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
            // 类型不匹配的比较
            _ => return Err(Error::query_execution("类型不匹配的比较".to_string())),
        };

        let result = match operator {
            ComparisonOperator::Equal => comparison == Ordering::Equal,
            ComparisonOperator::NotEqual => comparison != Ordering::Equal,
            ComparisonOperator::LessThan => comparison == Ordering::Less,
            ComparisonOperator::LessThanOrEqual => comparison != Ordering::Greater,
            ComparisonOperator::GreaterThan => comparison == Ordering::Greater,
            ComparisonOperator::GreaterThanOrEqual => comparison != Ordering::Less,
            ComparisonOperator::Like => {
                // 简化的LIKE实现
                if let (Value::Text(text), Value::Text(pattern)) = (left, right) {
                    // 简单的通配符匹配
                    if pattern.contains('%') {
                        let pattern_without_wildcard = pattern.replace('%', "");
                        text.contains(&pattern_without_wildcard)
                    } else {
                        text == pattern
                    }
                } else {
                    false
                }
            }
            ComparisonOperator::In => {
                // 简化的IN实现
                false // TODO: 实现IN操作符
            }
        };

        Ok(result)
    }
}

impl HashJoinOperator {
    /// 创建新的哈希连接算子
    pub fn new(
        left_input: Box<dyn Operator>,
        right_input: Box<dyn Operator>,
        join_condition: JoinCondition,
        join_type: JoinType,
    ) -> Self {
        let left_schema = left_input.schema();
        let right_schema = right_input.schema();
        let schema = left_schema.merge(right_schema);

        Self {
            left_input,
            right_input,
            join_condition,
            join_type,
            hash_table: None,
            schema,
            current_matches: Vec::new(),
            current_match_index: 0,
            current_right_tuple: None,
            statistics: OperatorStatistics::default(),
            is_open: false,
        }
    }
}

impl Operator for HashJoinOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        let start_time = Instant::now();

        // 打开输入算子
        self.left_input.open(context)?;
        self.right_input.open(context)?;

        // 构建哈希表（使用左侧输入）
        self.build_hash_table(context)?;

        self.is_open = true;
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

        tracing::debug!("哈希连接算子已打开");
        Ok(())
    }

    fn next(&mut self, context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("算子未打开".to_string()));
        }

        let start_time = Instant::now();

        loop {
            // 如果有当前匹配的元组，返回下一个匹配
            if self.current_match_index < self.current_matches.len() {
                if let Some(ref right_tuple) = self.current_right_tuple {
                    let left_tuple = &self.current_matches[self.current_match_index];
                    let joined_tuple = self.join_tuples(left_tuple, right_tuple)?;

                    self.current_match_index += 1;
                    self.statistics.tuples_output += 1;
                    self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

                    return Ok(Some(joined_tuple));
                }
            }

            // 获取下一个右侧元组
            if let Some(right_tuple) = self.right_input.next(context)? {
                self.statistics.tuples_processed += 1;

                // 在哈希表中查找匹配
                let hash_key = self.extract_hash_key(&right_tuple, self.join_condition.right_column)?;

                if let Some(hash_table) = &self.hash_table {
                    if let Some(matches) = hash_table.get(&hash_key) {
                        self.current_matches = matches.clone();
                        self.current_match_index = 0;
                        self.current_right_tuple = Some(right_tuple);

                        // 继续循环处理匹配
                        continue;
                    }
                }

                // 没有匹配，根据连接类型处理
                match self.join_type {
                    JoinType::Inner => {
                        // 内连接：没有匹配则跳过
                        continue;
                    }
                    JoinType::LeftOuter => {
                        // 左外连接：右侧没有匹配，返回NULL值
                        // TODO: 实现左外连接逻辑
                        continue;
                    }
                    _ => {
                        // 其他连接类型暂不支持
                        return Err(Error::query_execution("不支持的连接类型".to_string()));
                    }
                }
            } else {
                // 没有更多右侧元组
                self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
                return Ok(None);
            }
        }
    }

    fn close(&mut self, context: &mut ExecutionContext) -> Result<()> {
        self.left_input.close(context)?;
        self.right_input.close(context)?;
        self.hash_table = None;
        self.current_matches.clear();
        self.current_right_tuple = None;
        self.is_open = false;

        tracing::debug!("哈希连接算子已关闭");
        Ok(())
    }

    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }

    fn statistics(&self) -> OperatorStatistics {
        let mut stats = self.statistics.clone();
        let left_stats = self.left_input.statistics();
        let right_stats = self.right_input.statistics();

        stats.disk_io_count += left_stats.disk_io_count + right_stats.disk_io_count;
        stats.memory_usage_bytes += left_stats.memory_usage_bytes + right_stats.memory_usage_bytes;

        stats
    }

    fn reset(&mut self) -> Result<()> {
        self.left_input.reset()?;
        self.right_input.reset()?;
        self.hash_table = None;
        self.current_matches.clear();
        self.current_match_index = 0;
        self.current_right_tuple = None;
        self.statistics = OperatorStatistics::default();
        self.is_open = false;
        Ok(())
    }
}

impl HashJoinOperator {
    /// 构建哈希表
    fn build_hash_table(&mut self, context: &mut ExecutionContext) -> Result<()> {
        let mut hash_table: HashMap<Vec<u8>, Vec<ExecutionTuple>> = HashMap::new();

        // 遍历左侧输入构建哈希表
        while let Some(left_tuple) = self.left_input.next(context)? {
            let hash_key = self.extract_hash_key(&left_tuple, self.join_condition.left_column)?;

            hash_table.entry(hash_key)
                .or_default()
                .push(left_tuple);
        }

        // 估算内存使用量
        let estimated_memory = hash_table.len() * 64; // 简化估算
        context.allocate_memory(estimated_memory)?;
        self.statistics.memory_usage_bytes += estimated_memory as u64;

        self.hash_table = Some(hash_table);

        tracing::debug!("哈希表构建完成，包含 {} 个桶", self.hash_table.as_ref().unwrap().len());
        Ok(())
    }

    /// 提取哈希键
    fn extract_hash_key(&self, tuple: &ExecutionTuple, column_index: usize) -> Result<Vec<u8>> {
        let value = tuple.get_value(column_index)?;

        // 将值序列化为字节数组作为哈希键
        let key = match value {
            Value::Integer(i) => i.to_le_bytes().to_vec(),
            Value::Real(f) => f.to_le_bytes().to_vec(),
            Value::Text(s) => s.as_bytes().to_vec(),
            Value::Boolean(b) => vec![if *b { 1 } else { 0 }],
            Value::Timestamp(ts) => ts.timestamp().to_le_bytes().to_vec(),
            Value::Null => vec![0], // NULL值的特殊键
            Value::Decimal(d) => d.to_string().into_bytes(), // 小数转字符串
        };

        Ok(key)
    }

    /// 连接两个元组
    fn join_tuples(&self, left_tuple: &ExecutionTuple, right_tuple: &ExecutionTuple) -> Result<ExecutionTuple> {
        let mut joined_values = left_tuple.values.clone();
        joined_values.extend(right_tuple.values.clone());

        // 保留左侧元组的行ID和版本信息
        Ok(ExecutionTuple {
            values: joined_values,
            row_id: left_tuple.row_id,
            version_info: left_tuple.version_info.clone(),
        })
    }
}

/// 聚合函数类型
#[derive(Debug, Clone, PartialEq)]
pub enum AggregateFunction {
    /// 计数
    Count,
    /// 求和
    Sum,
    /// 平均值
    Avg,
    /// 最小值
    Min,
    /// 最大值
    Max,
}

/// 聚合表达式
#[derive(Debug, Clone)]
pub struct AggregateExpression {
    /// 聚合函数
    pub function: AggregateFunction,
    /// 列索引（None表示COUNT(*)）
    pub column_index: Option<usize>,
    /// 输出列名
    pub output_name: String,
}

/// 分组键
type GroupKey = Vec<u8>;

/// 聚合状态
#[derive(Debug, Clone)]
struct AggregateState {
    /// 计数
    count: u64,
    /// 求和（用于Sum和Avg）
    sum: f64,
    /// 最小值
    min: Option<Value>,
    /// 最大值
    max: Option<Value>,
}

impl AggregateState {
    fn new() -> Self {
        Self {
            count: 0,
            sum: 0.0,
            min: None,
            max: None,
        }
    }

    /// 更新聚合状态
    fn update(&mut self, value: &Value) -> Result<()> {
        self.count += 1;

        match value {
            Value::Integer(i) => {
                self.sum += *i as f64;
                self.update_min_max(value);
            }
            Value::Real(f) => {
                self.sum += f;
                self.update_min_max(value);
            }
            Value::Null => {
                // NULL值不参与聚合计算（除了COUNT）
            }
            _ => {
                self.update_min_max(value);
            }
        }

        Ok(())
    }

    fn update_min_max(&mut self, value: &Value) {
        if !matches!(value, Value::Null) {
            if self.min.is_none() || self.compare_values(value, self.min.as_ref().unwrap()) < 0 {
                self.min = Some(value.clone());
            }
            if self.max.is_none() || self.compare_values(value, self.max.as_ref().unwrap()) > 0 {
                self.max = Some(value.clone());
            }
        }
    }

    fn compare_values(&self, a: &Value, b: &Value) -> i32 {
        match (a, b) {
            (Value::Integer(a), Value::Integer(b)) => a.cmp(b) as i32,
            (Value::Real(a), Value::Real(b)) => a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal) as i32,
            (Value::Text(a), Value::Text(b)) => a.cmp(b) as i32,
            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b) as i32,
            (Value::Timestamp(a), Value::Timestamp(b)) => a.cmp(b) as i32,
            _ => 0, // 不同类型认为相等
        }
    }

    /// 获取聚合结果
    fn get_result(&self, function: &AggregateFunction) -> Value {
        match function {
            AggregateFunction::Count => Value::Integer(self.count as i64),
            AggregateFunction::Sum => Value::Real(self.sum),
            AggregateFunction::Avg => {
                if self.count > 0 {
                    Value::Real(self.sum / self.count as f64)
                } else {
                    Value::Null
                }
            }
            AggregateFunction::Min => self.min.clone().unwrap_or(Value::Null),
            AggregateFunction::Max => self.max.clone().unwrap_or(Value::Null),
        }
    }
}

/// 聚合算子
/// 实现分组聚合功能
pub struct AggregationOperator {
    /// 输入算子
    input: Box<dyn Operator>,
    /// 分组列索引
    group_by_columns: Vec<usize>,
    /// 聚合表达式
    aggregate_expressions: Vec<AggregateExpression>,
    /// 输出模式
    schema: ExecutionSchema,
    /// 聚合结果（分组键 -> 聚合状态）
    aggregation_results: HashMap<GroupKey, Vec<AggregateState>>,
    /// 结果迭代器
    result_iterator: Option<std::vec::IntoIter<ExecutionTuple>>,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
    /// 是否已完成聚合
    aggregation_done: bool,
}

impl AggregationOperator {
    /// 创建新的聚合算子
    pub fn new(
        input: Box<dyn Operator>,
        group_by_columns: Vec<usize>,
        aggregate_expressions: Vec<AggregateExpression>,
        schema: ExecutionSchema,
    ) -> Self {
        Self {
            input,
            group_by_columns,
            aggregate_expressions,
            schema,
            aggregation_results: HashMap::new(),
            result_iterator: None,
            statistics: OperatorStatistics::default(),
            is_open: false,
            aggregation_done: false,
        }
    }

    /// 生成分组键
    fn generate_group_key(&self, tuple: &ExecutionTuple) -> Result<GroupKey> {
        let mut key = Vec::new();

        for &column_index in &self.group_by_columns {
            let value = tuple.get_value(column_index)?;
            let value_bytes = self.serialize_value_for_key(value)?;
            key.extend(value_bytes);
            key.push(0); // 分隔符
        }

        Ok(key)
    }

    /// 将值序列化为键的字节表示
    fn serialize_value_for_key(&self, value: &Value) -> Result<Vec<u8>> {
        let bytes = match value {
            Value::Integer(i) => i.to_le_bytes().to_vec(),
            Value::Real(f) => f.to_le_bytes().to_vec(),
            Value::Text(s) => s.as_bytes().to_vec(),
            Value::Boolean(b) => vec![if *b { 1 } else { 0 }],
            Value::Timestamp(ts) => ts.timestamp().to_le_bytes().to_vec(),
            Value::Null => vec![255], // NULL值的特殊标记
            Value::Decimal(d) => d.to_string().into_bytes(), // 小数转字符串
        };
        Ok(bytes)
    }

    /// 执行聚合计算
    fn perform_aggregation(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.aggregation_done {
            return Ok(());
        }

        // 打开输入算子
        self.input.open(context)?;

        // 处理所有输入元组
        while let Some(tuple) = self.input.next(context)? {
            let group_key = self.generate_group_key(&tuple)?;

            // 获取或创建该分组的聚合状态
            let states = self.aggregation_results
                .entry(group_key)
                .or_insert_with(|| {
                    self.aggregate_expressions
                        .iter()
                        .map(|_| AggregateState::new())
                        .collect()
                });

            // 更新每个聚合表达式的状态
            for (i, expr) in self.aggregate_expressions.iter().enumerate() {
                let value = if let Some(column_index) = expr.column_index {
                    tuple.get_value(column_index)?.clone()
                } else {
                    // COUNT(*) 情况
                    Value::Integer(1)
                };

                states[i].update(&value)?;
            }
        }

        // 关闭输入算子
        self.input.close(context)?;
        self.aggregation_done = true;

        Ok(())
    }

    /// 生成结果元组
    fn generate_results(&mut self) -> Result<Vec<ExecutionTuple>> {
        let mut results = Vec::new();

        for (group_key, states) in &self.aggregation_results {
            let mut values = Vec::new();

            // 添加分组列的值
            // 注意：这里简化处理，实际应该从group_key反序列化
            // 为了测试，我们先跳过分组列的值重建
            for _ in &self.group_by_columns {
                values.push(Value::Text("group_value".to_string())); // 占位符
            }

            // 添加聚合结果
            for (i, expr) in self.aggregate_expressions.iter().enumerate() {
                let result = states[i].get_result(&expr.function);
                values.push(result);
            }

            let tuple = ExecutionTuple {
                values,
                row_id: None,
                version_info: None,
            };

            results.push(tuple);
        }

        Ok(results)
    }
}

impl Operator for AggregationOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        let start_time = Instant::now();

        // 执行聚合计算
        self.perform_aggregation(context)?;

        // 生成结果
        let results = self.generate_results()?;
        self.result_iterator = Some(results.into_iter());

        self.is_open = true;
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

        tracing::debug!("聚合算子已打开，分组数量: {}", self.aggregation_results.len());
        Ok(())
    }

    fn next(&mut self, _context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("聚合算子未打开".to_string()));
        }

        let start_time = Instant::now();

        let result = if let Some(ref mut iterator) = self.result_iterator {
            iterator.next()
        } else {
            None
        };

        if result.is_some() {
            self.statistics.tuples_output += 1;
        }

        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
        Ok(result)
    }

    fn close(&mut self, _context: &mut ExecutionContext) -> Result<()> {
        self.is_open = false;
        self.result_iterator = None;
        tracing::debug!("聚合算子已关闭");
        Ok(())
    }

    fn reset(&mut self) -> Result<()> {
        self.is_open = false;
        self.aggregation_done = false;
        self.aggregation_results.clear();
        self.result_iterator = None;
        self.statistics = OperatorStatistics::default();
        self.input.reset()?;
        Ok(())
    }

    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }

    fn statistics(&self) -> OperatorStatistics {
        self.statistics.clone()
    }
}

/// 排序方向
#[derive(Debug, Clone, PartialEq)]
pub enum SortDirection {
    /// 升序
    Ascending,
    /// 降序
    Descending,
}

/// 排序键
#[derive(Debug, Clone)]
pub struct SortKey {
    /// 列索引
    pub column_index: usize,
    /// 排序方向
    pub direction: SortDirection,
}

/// 排序算子
/// 对输入数据进行排序
pub struct SortOperator {
    /// 输入算子
    input: Box<dyn Operator>,
    /// 排序键列表
    sort_keys: Vec<SortKey>,
    /// 输出模式（与输入相同）
    schema: ExecutionSchema,
    /// 排序后的结果
    sorted_results: Vec<ExecutionTuple>,
    /// 结果迭代器
    result_iterator: Option<std::vec::IntoIter<ExecutionTuple>>,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
    /// 是否已完成排序
    sorting_done: bool,
}

impl SortOperator {
    /// 创建新的排序算子
    pub fn new(
        input: Box<dyn Operator>,
        sort_keys: Vec<SortKey>,
        schema: ExecutionSchema,
    ) -> Self {
        Self {
            input,
            sort_keys,
            schema,
            sorted_results: Vec::new(),
            result_iterator: None,
            statistics: OperatorStatistics::default(),
            is_open: false,
            sorting_done: false,
        }
    }

    /// 执行排序
    fn perform_sort(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.sorting_done {
            return Ok(());
        }

        // 打开输入算子
        self.input.open(context)?;

        // 收集所有输入元组
        let mut tuples = Vec::new();
        while let Some(tuple) = self.input.next(context)? {
            tuples.push(tuple);
        }

        // 关闭输入算子
        self.input.close(context)?;

        // 执行排序
        tuples.sort_by(|a, b| self.compare_tuples(a, b));

        self.sorted_results = tuples;
        self.sorting_done = true;

        Ok(())
    }

    /// 比较两个元组
    fn compare_tuples(&self, a: &ExecutionTuple, b: &ExecutionTuple) -> std::cmp::Ordering {
        for sort_key in &self.sort_keys {
            let value_a = a.get_value(sort_key.column_index);
            let value_b = b.get_value(sort_key.column_index);

            let ordering = match (value_a, value_b) {
                (Ok(val_a), Ok(val_b)) => self.compare_values(val_a, val_b),
                (Err(_), Ok(_)) => std::cmp::Ordering::Greater, // 错误值排在后面
                (Ok(_), Err(_)) => std::cmp::Ordering::Less,
                (Err(_), Err(_)) => std::cmp::Ordering::Equal,
            };

            let final_ordering = match sort_key.direction {
                SortDirection::Ascending => ordering,
                SortDirection::Descending => ordering.reverse(),
            };

            if final_ordering != std::cmp::Ordering::Equal {
                return final_ordering;
            }
        }

        std::cmp::Ordering::Equal
    }

    /// 比较两个值
    fn compare_values(&self, a: &Value, b: &Value) -> std::cmp::Ordering {
        match (a, b) {
            (Value::Null, Value::Null) => std::cmp::Ordering::Equal,
            (Value::Null, _) => std::cmp::Ordering::Less, // NULL值排在前面
            (_, Value::Null) => std::cmp::Ordering::Greater,
            (Value::Integer(a), Value::Integer(b)) => a.cmp(b),
            (Value::Real(a), Value::Real(b)) => a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal),
            (Value::Integer(a), Value::Real(b)) => (*a as f64).partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal),
            (Value::Real(a), Value::Integer(b)) => a.partial_cmp(&(*b as f64)).unwrap_or(std::cmp::Ordering::Equal),
            (Value::Text(a), Value::Text(b)) => a.cmp(b),
            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
            (Value::Timestamp(a), Value::Timestamp(b)) => a.cmp(b),
            _ => std::cmp::Ordering::Equal, // 不同类型认为相等
        }
    }
}

impl Operator for SortOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        let start_time = Instant::now();

        // 执行排序
        self.perform_sort(context)?;

        // 创建结果迭代器
        self.result_iterator = Some(self.sorted_results.clone().into_iter());

        self.is_open = true;
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

        tracing::debug!("排序算子已打开，排序元组数量: {}", self.sorted_results.len());
        Ok(())
    }

    fn next(&mut self, _context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("排序算子未打开".to_string()));
        }

        let start_time = Instant::now();

        let result = if let Some(ref mut iterator) = self.result_iterator {
            iterator.next()
        } else {
            None
        };

        if result.is_some() {
            self.statistics.tuples_output += 1;
        }

        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
        Ok(result)
    }

    fn close(&mut self, _context: &mut ExecutionContext) -> Result<()> {
        self.is_open = false;
        self.result_iterator = None;
        tracing::debug!("排序算子已关闭");
        Ok(())
    }

    fn reset(&mut self) -> Result<()> {
        self.is_open = false;
        self.sorting_done = false;
        self.sorted_results.clear();
        self.result_iterator = None;
        self.statistics = OperatorStatistics::default();
        self.input.reset()?;
        Ok(())
    }

    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }

    fn statistics(&self) -> OperatorStatistics {
        self.statistics.clone()
    }
}

/// 限制算子
/// 限制输出的元组数量
pub struct LimitOperator {
    /// 输入算子
    input: Box<dyn Operator>,
    /// 限制数量
    limit: usize,
    /// 偏移量（OFFSET）
    offset: usize,
    /// 输出模式（与输入相同）
    schema: ExecutionSchema,
    /// 当前输出计数
    output_count: usize,
    /// 当前跳过计数
    skip_count: usize,
    /// 统计信息
    statistics: OperatorStatistics,
    /// 是否已打开
    is_open: bool,
}

impl LimitOperator {
    /// 创建新的限制算子
    pub fn new(
        input: Box<dyn Operator>,
        limit: usize,
        offset: usize,
        schema: ExecutionSchema,
    ) -> Self {
        Self {
            input,
            limit,
            offset,
            schema,
            output_count: 0,
            skip_count: 0,
            statistics: OperatorStatistics::default(),
            is_open: false,
        }
    }

    /// 创建只有LIMIT的算子（无OFFSET）
    pub fn with_limit_only(
        input: Box<dyn Operator>,
        limit: usize,
        schema: ExecutionSchema,
    ) -> Self {
        Self::new(input, limit, 0, schema)
    }
}

impl Operator for LimitOperator {
    fn open(&mut self, context: &mut ExecutionContext) -> Result<()> {
        if self.is_open {
            return Ok(());
        }

        let start_time = Instant::now();

        // 打开输入算子
        self.input.open(context)?;

        self.is_open = true;
        self.output_count = 0;
        self.skip_count = 0;
        self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;

        tracing::debug!("限制算子已打开，限制: {}, 偏移: {}", self.limit, self.offset);
        Ok(())
    }

    fn next(&mut self, context: &mut ExecutionContext) -> Result<Option<ExecutionTuple>> {
        if !self.is_open {
            return Err(Error::query_execution("限制算子未打开".to_string()));
        }

        let start_time = Instant::now();

        // 如果已经输出了足够的元组，返回None
        if self.output_count >= self.limit {
            self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
            return Ok(None);
        }

        // 跳过OFFSET指定的元组
        while self.skip_count < self.offset {
            if (self.input.next(context)?).is_some() {
                self.skip_count += 1;
            } else {
                // 输入已结束
                self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
                return Ok(None);
            }
        }

        // 获取下一个元组
        if let Some(tuple) = self.input.next(context)? {
            self.output_count += 1;
            self.statistics.tuples_output += 1;
            self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
            Ok(Some(tuple))
        } else {
            self.statistics.execution_time_us += start_time.elapsed().as_micros() as u64;
            Ok(None)
        }
    }

    fn close(&mut self, context: &mut ExecutionContext) -> Result<()> {
        self.is_open = false;
        self.input.close(context)?;
        tracing::debug!("限制算子已关闭");
        Ok(())
    }

    fn reset(&mut self) -> Result<()> {
        self.is_open = false;
        self.output_count = 0;
        self.skip_count = 0;
        self.statistics = OperatorStatistics::default();
        self.input.reset()?;
        Ok(())
    }

    fn schema(&self) -> &ExecutionSchema {
        &self.schema
    }

    fn statistics(&self) -> OperatorStatistics {
        self.statistics.clone()
    }
}
